package cover;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

/**
 * Created by zhaopeng on 2/14/17.
 */
public class CoverReducer extends Reducer<Text, Text, Text, Text> {
    /**
     * Key: token, value: 当前token在当前reduce阶段中来自log中的token计数
     */
    //private Map<String, Integer> tokenMapper = new HashMap<>();

    /**
     * key: token, value: [token的识别量， token的未识别量]
     */
    private Map<String, List<Integer>> resultCounter = new HashMap<>();
    private Text writeKey = new Text();
    private Text writeValue = new Text();

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        int cover = 0, uncover = 0;
        for (Text value : values) {
            String tostring = value.toString();
            String[] numArray = tostring.split(Pattern.quote("\t"));
            cover += Integer.parseInt(numArray[0]);
            uncover += Integer.parseInt(numArray[1]);
        }

        if (resultCounter.containsKey(key.toString())) {
            List<Integer> list = resultCounter.get(key.toString());
            list.set(0, list.get(0) + cover);
            list.set(1, list.get(1) + uncover);
        } else {
            List<Integer> list = new ArrayList<>();
            list.add(0);
            list.add(0);
            list.set(0, cover);
            list.set(1, uncover);

            resultCounter.put(key.toString(),list);
        }
    }


    //writeValue.set(result);
    //context.write(key,writeValue);
        /*boolean isExisted = false;
        boolean isFromLog = false;
        tokenMapper.clear();

        for(Text value:values){
            if(value.toString().startsWith("token:")) {
                isFromLog = true;
                String toString = value.toString();
                String[] tokens =(toString.substring(6)).split(Pattern.quote(";"));
                for (String tokenValue : tokens) {
                    String[] tmparray = tokenValue.split(Pattern.quote(":"));
                    if ("true".equals(tmparray[0])) {
                        isExisted = true;
                    } else {
                        String ftoken = tmparray[0];
                        Integer num = Integer.parseInt(tmparray[1]);
                        if (tokenMapper.containsKey(ftoken)) {
                            tokenMapper.put(ftoken, tokenMapper.get(ftoken) + num);
                        } else {
                            tokenMapper.put(ftoken, num);
                        }
                    }
                }
            }else if ("true".equals(value.toString())){
                isExisted = true;
            }
        }
*/
        /*(for (Text value : values) {
            String data = value.toString();
            if (data.startsWith("token:")) { // From log
                isFromLog = true;
                String token = data.substring(6);
                if (!tokenMapper.containsKey(token)) {
                    tokenMapper.put(token, 1);
                } else {
                    tokenMapper.put(token, tokenMapper.get(token) + 1);
                }
            } else {
                isExisted = true;
            }
        }*/
       /* if (isFromLog) {
            for (String token : tokenMapper.keySet()) {
                if (!resultCounter.containsKey(token)) {
                    List<Integer> list = new ArrayList<>();
                    list.add(0);
                    list.add(0);
                    resultCounter.put(token, list);
                }
                List<Integer> list = resultCounter.get(token);
                if (isExisted) {
                    list.set(0, list.get(0) + tokenMapper.get(token));
                } else {
                    list.set(1, list.get(1) + tokenMapper.get(token));
                }
            }
        }

*/
    //        tokenMapper.clear();
//        boolean isLoguid = false;
//        boolean isCover = false;
//        for (Text tmp : values) {
//            String args = tmp.toString();
//            if (-1 != args.indexOf("token:")) {
//                isLoguid = true;
//                String token = args.split(Pattern.quote(":"))[1];
//                if (tokenMapper.containsKey(token)) {
//                    int num = tokenMapper.get(token) + 1;
//
//                    tokenMapper.put(token, num);
//                } else {
//                    tokenMapper.put(token, 1);
//                }
//            } else if (-1 != args.indexOf("true")) {
//                isCover = true;
//            }
//        }
//        if (isLoguid) {
//            String tokens = "";
//            Iterator it = tokenMapper.entrySet().iterator();
//            Map.Entry entry;
//            if (it.hasNext()) {
//                entry=(Map.Entry)it.next();
//                tokens += entry.getKey().toString() + ":" + entry.getValue();
//            }
//            while (it.hasNext()) {
//                entry = (Map.Entry)it.next();
//                tokens += "," + entry.getKey().toString() + ":" + entry.getValue();
//            }
//
//            tokens += ";" + isCover;
//            value.set(tokens);
//            context.write(key, value);
//        }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for (String token : resultCounter.keySet()) {
            List<Integer> list = resultCounter.get(token);
            writeKey.set(token);
            writeValue.set(list.get(0) + "\t" + list.get(1));
            context.write(writeKey, writeValue);
        }
    }
}
